package com.study.crypto.basic.kafka.admin.controller;

import com.study.crypto.basic.kafka.admin.component.KafkaConfig;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;

/**
 * @author Songjin
 * @since 2021-09-26 17:40
 */
@RestController
public class KafkaTopicController {
    
    @Autowired
    private KafkaConfig kafkaConfig;
    
    @GetMapping("/createTopic")
    public void createTopic() {
        AdminClient adminClient = AdminClient.create(kafkaConfig.getProps());
    
        NewTopic newTopic = new NewTopic("test2", 1, (short) 1);
        List<NewTopic> newTopics = Collections.singletonList(newTopic);
        CreateTopicsResult topics = adminClient.createTopics(newTopics);
        System.out.println("topics.all().isDone():  " + topics.all().isDone());
    
        adminClient.close();
    }
    
    @GetMapping("/deleteTopic")
    public void deleteTopic() {
        AdminClient adminClient = AdminClient.create(kafkaConfig.getProps());
        adminClient.deleteTopics(Collections.singletonList("test1"));
        adminClient.close();
    }
    
    @GetMapping("/listAllTopic")
    public void listAllTopic() {
        AdminClient              adminClient = AdminClient.create(kafkaConfig.getProps());
        ListTopicsResult         result      = adminClient.listTopics();
        KafkaFuture<Set<String>> names       = result.names();
        try {
            DescribeClusterResult    describeClusterResult = adminClient.describeCluster();
            System.out.println(describeClusterResult.clusterId().get());
            names.get().forEach(k -> System.out.println(k));
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        adminClient.close();
    }
    
    @GetMapping("getTopic")
    public void getTopic() {
        AdminClient adminClient = AdminClient.create(kafkaConfig.getProps());
        
        DescribeTopicsResult describeTopics = adminClient.describeTopics(Collections.singletonList("syn-test"));
        
        Collection<KafkaFuture<TopicDescription>> values = describeTopics.values().values();
        
        if (values.isEmpty()) {
            System.out.println("找不到描述信息");
        } else {
            for (KafkaFuture<TopicDescription> value : values) {
                System.out.println(value);
            }
        }
        adminClient.close();
    }
}